package com.techsun.datanow.sync

import com.microsoft.azure.cosmosdb.spark.config.Config
import com.microsoft.azure.cosmosdb.spark.schema._
import org.apache.kudu.spark.kudu.KuduContext
import org.apache.log4j.{Level, Logger}
import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.types._


class Sync_OrderPayment extends TDataSync {
  /**
   * 同步工作方法
   *
   * @param spark             spark 会话对象
   * @param cosmosDbEndpoint  CosmosDB Endpoint
   * @param cosmosDbMasterkey CosmosDB Endpoint 对应的密钥
   * @param kuduMaster        kudu Master 节点
   * @return 同步成功或失败
   */
  override def sync(spark: SparkSession, cosmosDbEndpoint: String, cosmosDbMasterkey: String, kuduMaster: String, beginTime: Long, endTime: Long): Boolean = {
    try {
      Logger.getLogger("org").setLevel(Level.ERROR)
      Logger.getLogger("com").setLevel(Level.ERROR)
      val kuduContext = new KuduContext(kuduMaster, spark.sparkContext)
      val tableName = "sh_tsrorderpayment"
      val config = Config(Map("Endpoint" -> cosmosDbEndpoint,
        "Masterkey" -> cosmosDbMasterkey,
        "Database" -> "VF_Vans_CN",
        "preferredRegions" -> "China East 2;",
        "Collection" -> "TsrOrderPayment",
        "query_custom" ->
          s""" SELECT c.id id, c.name name, c.orderId orderid, c.paymentId paymentid, c.paymentCode paymentcode, c.paymentName paymentname, c.payAmount as payamount, c.totalAmount as totalamount, c.costAmount as costamount, c.couponId as couponid, c.createTime as createtime, c.updateTime as updatetime, c._ts as _ts
             |FROM c
             | where c._ts >= ${beginTime}""".stripMargin))

      val schema = StructType(
        List(
          StructField("id", StringType, true),
          StructField("orderid", StringType, true),
          StructField("productid", StringType, true),
          StructField("productcode", StringType, true),
          StructField("price", DoubleType, true),
          StructField("totalamount", DoubleType, true),
          StructField("couponamount", DoubleType, true),
          StructField("discountamount", DoubleType, true),
          StructField("totaldisamount", DoubleType, true),
          StructField("costamount", DoubleType, true),
          StructField("grossamount", DoubleType, true),
          StructField("scoreamount", DoubleType, true),
          StructField("qty", IntegerType, true),
          StructField("title", StringType, true),
          StructField("createtime", TimestampType, true),
          StructField("updatetime", TimestampType, true),
          StructField("_ts", IntegerType, true)
        ))
      val df = spark.sqlContext.read.schema(schema).cosmosDB(config)
      val count = df.count();
      printf("Sync_TsrOrderPayment query count: %d\n", count)
      kuduContext.upsertRows(df, tableName)
      return true
    } catch {
      case ex: Exception => {
        println("Sync_TsrOrderPayment exception: " + ex.getMessage)
        return false
      }
    }
  }
}
